package com.scut.chenlujie.library.clink.impl.async;

import com.scut.chenlujie.library.clink.core.IoArgs;
import com.scut.chenlujie.library.clink.core.ReceiveDispatcher;
import com.scut.chenlujie.library.clink.core.ReceivePacket;
import com.scut.chenlujie.library.clink.core.Receiver;
import com.scut.chenlujie.library.clink.utils.CloseUtils;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;

public class AsyncReceiveDispatcher implements ReceiveDispatcher,
        IoArgs.IoArgsEventProcessor,
        AsyncPacketWriter.PacketProvider {
    private final AtomicBoolean isClose = new AtomicBoolean(false);
    private final Receiver receiver;
    private final ReceivePacketCallback receivePacketCallback;

    private final AsyncPacketWriter writer = new AsyncPacketWriter(this);

    public AsyncReceiveDispatcher(Receiver receiver, ReceivePacketCallback receivePacketCallback) {
        this.receiver = receiver;
        this.receiver.setReceiveListener(this);
        this.receivePacketCallback = receivePacketCallback;
    }

    @Override
    public void start() {
        registerReceive();
    }

    private void registerReceive() {
        try {
            receiver.postReceiveAsync();
        } catch (IOException e) {
            closeAndNotify();
        }
    }

    private void closeAndNotify() {
        CloseUtils.close(this);
    }

    @Override
    public void stop() {

    }

    @Override
    public void close() throws IOException {
        if (isClose.compareAndSet(false, true)) {
            writer.close();
        }
    }

    @Override
    public IoArgs provideIoArgs() {
        IoArgs args =  writer.takeIoArgs();
        //一份新的IoArgs需要调用一次k开始写入数据的操作
        args.startWriting();
        return args;
    }

    @Override
    public void onConsumeFailed(IoArgs args, Exception e) {
        e.printStackTrace();
    }


    //IoArgs从channel中去读完数据触发
    @Override
    public void onConsumeCompleted(IoArgs args) {
        if(isClose.get()){
            return;
        }

        //消费数据之前标识它已经完成
        //改变未可读数据状态
        args.finishWriting();

        //有数据则重复消费
        do {
            //把Args中的数据写入帧
            writer.consumeIoArgs(args);
        } while (args.remained() && !isClose.get());
        registerReceive();
    }

    @Override
    public ReceivePacket takePacket(byte type, long length, byte[] headerInfo) {
        return receivePacketCallback.onArrivedNewPacket(type, length);
    }

    @Override
    public void completedPacket(ReceivePacket packet, boolean isSucceed) {
        CloseUtils.close(packet);
        receivePacketCallback.onReceivePacketCompleted(packet);
    }

    @Override
    public void onReceiveHeartbeat() {
        receivePacketCallback.onArrivedHeartbeat();
    }
}

